package com.autumn.redis;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.TimeoutUtils;

import com.autumn.util.function.FunctionOneResult;
import com.autumn.util.tuple.TupleTwo;

/**
 * 值操作
 * 
 * @author 老码农
 *         <p>
 *         Description
 *         </p>
 * @date 2017-12-13 01:13:17
 * @param <T>
 */
class AutumnValueOperations<T> extends AbstractOperations {
	private final AutumnRedisTemplate template;

	/**
	 * 
	 * @param template
	 */
	public AutumnValueOperations(AutumnRedisTemplate template) {
		this.template = template;
	}

	/**
	 * 设置
	 * 
	 * @param key
	 *            键值
	 * @param value
	 * @return
	 */
	public void set(String key, T value) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		final byte[] rawValue = AutumnRedisTemplate.STRING_SERIALIZER
				.serialize(value == null ? null : value.toString());
		template.execute(new RedisCallback<T>() {
			@Override
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				connection.set(rawKey, rawValue);
				return value;
			}
		}, true);
	}

	/**
	 * 设置
	 * 
	 * @param key
	 *            键
	 * @param value
	 *            值
	 * @param timeout
	 *            超时时间
	 * @param unit
	 *            时间单位
	 */
	public void set(String key, T value, long timeout, TimeUnit unit) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		final byte[] rawValue = AutumnRedisTemplate.STRING_SERIALIZER
				.serialize(value == null ? null : value.toString());
		template.execute(new RedisCallback<T>() {
			@Override
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				potentiallyUsePsetEx(connection);
				return value;
			}

			private void potentiallyUsePsetEx(RedisConnection connection) {
				if (!TimeUnit.MILLISECONDS.equals(unit) || !failsafeInvokePsetEx(connection)) {
					connection.setEx(rawKey, TimeoutUtils.toSeconds(timeout, unit), rawValue);
				}
			}

			private boolean failsafeInvokePsetEx(RedisConnection connection) {
				boolean failed = false;
				try {
					connection.pSetEx(rawKey, timeout, rawValue);
				} catch (UnsupportedOperationException e) {
					// in case the connection does not support pSetEx return
					// false to allow fallback to other operation.
					failed = true;
				}
				return !failed;
			}

		}, true);
	}

	/**
	 * 获取
	 * 
	 * @param key
	 *            键
	 * @param func
	 *            函数转换
	 * @return
	 */
	public T get(String key, FunctionOneResult<String, T> func) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		return template.execute(new RedisCallback<T>() {
			@Override
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				String value = AutumnRedisTemplate.STRING_SERIALIZER.deserialize(connection.get(rawKey));
				return func.apply(value);
			}
		}, true);
	}

	/**
	 * 递增(++)
	 * 
	 * @param key
	 *            键
	 * @return
	 */
	public Long incr(String key) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		return template.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.incr(rawKey);
			}
		}, true);
	}

	/**
	 * 递增(+=value)
	 * 
	 * @param key
	 *            键
	 * @return
	 */
	public Long incrBy(String key, long delta) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		return template.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.incrBy(rawKey, delta);
			}
		}, true);
	}

	/**
	 * 递减(--)
	 * 
	 * @param key
	 *            键
	 * @return
	 */
	public Long decr(String key) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		return template.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.decr(rawKey);
			}
		}, true);
	}

	/**
	 * 递减(-=value)
	 * 
	 * @param key
	 *            键
	 * @return
	 */
	public Long decrBy(String key, long delta) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		return template.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				return connection.incrBy(rawKey, delta);
			}
		}, true);
	}

	private final static int DECR_BY_IF_DATA_SIZE = 2;
	
	/**
	 * 条件递减
	 * 
	 * @param key
	 *            键
	 * @param delta
	 *            步长
	 * @param minValue
	 *            最小值
	 * @return
	 */
	public TupleTwo<Boolean, Long> decrByIf(String key, long delta, long minValue) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		final byte[] rawdelta = AutumnRedisTemplate.STRING_SERIALIZER.serialize(Long.toString(delta));
		final byte[] rawMinValue = AutumnRedisTemplate.STRING_SERIALIZER.serialize(Long.toString(minValue));
		return template.execute(new RedisCallback<TupleTwo<Boolean, Long>>() {
			@Override
			public TupleTwo<Boolean, Long> doInRedis(RedisConnection connection) throws DataAccessException {
				List<Object> resultBytes = connection.eval(SCRIPT_DECRBY_MIN_VALUE_BYTE, ReturnType.MULTI, 1, rawKey,
						rawMinValue, rawdelta);
				if (resultBytes != null && resultBytes.size() == DECR_BY_IF_DATA_SIZE) {
					boolean isSuccess = (long) resultBytes.get(0) > 0L;
					Object result = resultBytes.get(1);
					if (result == null) {
						return new TupleTwo<Boolean, Long>(isSuccess, null);
					}
					if (result instanceof Long) {
						return new TupleTwo<Boolean, Long>(isSuccess, (Long) result);
					}
					String str = AutumnRedisTemplate.STRING_SERIALIZER.deserialize((byte[]) result);
					if (str == null) {
						return new TupleTwo<Boolean, Long>(isSuccess, null);
					}
					return new TupleTwo<Boolean, Long>(isSuccess, Long.parseLong(str));
				}
				return new TupleTwo<Boolean, Long>(true, 0L);
			}
		}, true);
	}

	/**
	 * 获取后删除
	 * 
	 * @param key
	 *            键
	 * @param func
	 *            函数转换
	 * @return
	 */
	public T getByDelete(String key, FunctionOneResult<String, T> func) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(key);
		return template.execute(new RedisCallback<T>() {
			@Override
			public T doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] result = connection.eval(SCRIPT_READ_DELETE_BYTE, ReturnType.VALUE, 1, rawKey);
				String value = AutumnRedisTemplate.STRING_SERIALIZER.deserialize(result);
				return func.apply(value);
			}
		}, true);
	}

	/**
	 * 删除匹配
	 * 
	 * @param keyExpression
	 *            键表达式
	 *            <p>
	 *            a* 表示a开头的键
	 *            </p>
	 *            <p>
	 *            *a 表示对a结束的键
	 *            </p>
	 * @return 返回成功删除的键数量
	 */
	public long deleteByMatches(String keyExpression) {
		final byte[] rawKey = AutumnRedisTemplate.STRING_SERIALIZER.serialize(keyExpression);
		return template.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) throws DataAccessException {
				Long value = connection.eval(SCRIPT_MATCHES_DELETE_BYTE, ReturnType.INTEGER, 0, rawKey);
				return value;
			}
		}, true);
	}
}